package cn.sheep.dolphin.report

import cn.sheep.dolphin.common.DolphinAppComm
import cn.sheep.dolphin.config.DolphinConfig
import org.apache.spark.sql.{SQLContext, SaveMode}

/** 地域指标统计 - sql
  * author: old sheep
  * QQ: 64341393 
  * Created 2018/11/29
  */
object AreaAnalysisSQL {

	def main(args: Array[String]): Unit = {

		if (args.size != 1) {
			println(
				"""
				  |Usage: cn.sheep.dolphin.report.AreaAnalysisSQL
				  |	<parquetInputPath>
				""".stripMargin)
			sys.exit(-1)
		}

		val Array(parquetInputPath) = args

		val sc = DolphinAppComm.createSparkContext("地域指标统计")
		val sQLContext = new SQLContext(sc)

		// 读取数据parquet
		val dataFrame = sQLContext.read.parquet(parquetInputPath)

		// 将dataFrame注册成临时表
		dataFrame.registerTempTable("adlogs")

		// spark sql提供了udf功能
		sQLContext.udf.register("sheepif", (bool: Boolean, right: Double, wrong: Double) => if (bool) right else wrong)

		// 写sql查表
		val result = sQLContext.sql(
			"""
			  |select provincename, cityname, substring(requestdate, 0, 10) day,
			  |sum(case when requestmode=1 and processnode>=1 then 1 else 0 end) rawReq,
			  |sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) effReq,
			  |sum(case when requestmode=1 and processnode=3 then 1 else 0 end) adReq,
			  |
			  |sum(sheepif(iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0, 1, 0)) isRTB,
			  |sum(if(iseffective=1 and isbilling=1 and iswin=1, 1, 0)) succRTB,
			  |
			  |sum(if(requestmode=2 and iseffective=1, 1, 0)) adShow,
			  |sum(if(requestmode=3 and iseffective=1, 1, 0)) adClik,
			  |
			  |sum(if(iseffective=1 and isbilling=1 and iswin=1, winprice/1000, 0)) expense,
			  |sum(if(iseffective=1 and isbilling=1 and iswin=1, adpayment/1000, 0)) cost
			  |
			  |from adlogs
			  |group by provincename, cityname, substring(requestdate, 0, 10)
			  |
			""".stripMargin)

		// 将结果写入到数据库中
		result.write.mode(SaveMode.Overwrite).jdbc(DolphinConfig._url, "r_area_analysis2", DolphinConfig.props)


		sc.stop()
	}

}
